<!DOCTYPE html>
<html>
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>RxJS Overview 阅读笔记</title>
  <link rel="icon" type="image/png" href="../think/assets/ant.png">
  <link href="../styles.css" rel="stylesheet">
  <link href="../vs.css" rel="stylesheet">
  <!-- Global site tag (gtag.js) - Google Analytics -->
  <script async src="https://www.googletagmanager.com/gtag/js?id=UA-131906119-1"></script>
  <script>
    window.dataLayer = window.dataLayer || [];
    function gtag(){dataLayer.push(arguments);}
    gtag('js', new Date());

    gtag('config', 'UA-131906119-1');
  </script>
  <script async src="../index.js"></script>
</head>
<body>
<h1><em>RxJS Overview</em> 阅读笔记</h1>
<p><em>RxJS Overview</em>为官方对RxJS的一个概述，点击<a href="http://reactivex.io/rxjs/manual/overview.html">这里</a>进行原文阅读。</p>
<p>非常感谢 @雪狼 @trotyl 指出我在概念理解上的问题。</p>
<h2>前言</h2>
<p>本文是我阅读<em>RxJS Overview</em>时所做的笔记。该概述（Overview）写得非常之清晰和简洁，是很好的入门RxJS资料。概述内容包括：<strong>RxJS涉及到的主要概念及概念相关的必要背景知识</strong>，其中主要概念包括：Observable、Observer、Subscription、Subject、Operator、Scheduler，本文会一一进行介绍。尤其是那些时不时出现的手书斜体格式的总结文字，非常值得认真思考。本文尝试进行总结，力求把RxJS的核心思想表达清楚。</p>
<p>用一句话介绍RxJS的话，就是<strong>RxJS将异步数据抽象为数据流，并提供对数据流进行各种转换操作的接口</strong>。</p>
<h2>数据传输</h2>
<p>数据从生产者传输到消费者那里，有两种方式：一是消费者从生产者那里拉取（<strong>pull</strong>）数据，二是生产者推送（<strong>push</strong>）数据给消费者。前者就是常见的函数调用（消费者调用生产数据的函数），后者则是用于处理异步的回调方式（消费者只定义接受数据的函数，然后供生产者调用）。它们的区别在于由哪一方去主动调用。代码如下：</p>
<pre><code class="language-js"><span class="hljs-comment">// 拉取式</span>
<span class="hljs-keyword">var</span> producer = <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params"></span>) </span>{
    <span class="hljs-keyword">var</span> data = <span class="hljs-number">42</span>;
    <span class="hljs-keyword">return</span> data;
}
<span class="hljs-keyword">var</span> consumer = <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params"></span>) </span>{
    <span class="hljs-comment">// 数据消费者主动调用数据生产函数</span>
    <span class="hljs-keyword">var</span> data = producer();
    <span class="hljs-comment">// 可以消费数据了</span>
}
<span class="hljs-comment">// 直接获取数据并进行消费</span>
consumer();
</code></pre>
<pre><code class="language-js"><span class="hljs-comment">// 推送式</span>
<span class="hljs-keyword">var</span> producer = <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">consumer</span>) </span>{
    setTimeout(<span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params"></span>) </span>{
        <span class="hljs-keyword">var</span> data = <span class="hljs-number">42</span>;
        <span class="hljs-comment">// 数据生产者推送数据给数据消费者</span>
        consumer(data);
    }, <span class="hljs-number">3000</span>)
}
<span class="hljs-keyword">var</span> consumer = <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">data</span>) </span>{
    <span class="hljs-comment">// 生产者通过参数把数据推送给消费者函数</span>
}
<span class="hljs-comment">// 等到异步数据准备好时，就会调用消费者</span>
<span class="hljs-comment">// 调用之后才可以消费数据</span>
producer(consumer);
</code></pre>
<p>拉取方式常用在获取同步数据上，而推送方式则更多用在获取异步数据上。会有异步数据这个说法，是因为有些数据的获取需要等待（比如浏览器通过URL接口获取服务器上的数据）。如果数据消费者在原地傻傻地等着数据，而不去做其他的事情，必然导致效率低下。正确的做法是在数据还没有到的时候去做其他的事情，等数据到了再来通知消费者。</p>
<p>在获取异步数据上面，常见的有事件监听和Promise（都是通过回调）。区别是事件监听可以触发多次回调，一个Promise只能触发一次回调。<strong>一次和多次在更深层次上的区别，是生产者函数可以有多少个返回值</strong>。RxJS是两者的结合，<strong>提供了Promise形式的异步编程方式（看起来像同步），但又可以有多个返回值</strong>。对比以下代码就很容易理解上一句话了：</p>
<pre><code class="language-js"><span class="hljs-comment">// 事件监听</span>
button.addEventListener(<span class="hljs-string">'click'</span>, <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">event</span>) </span>{
    <span class="hljs-comment">// 触发回调的时机是该元素被点击了</span>
    <span class="hljs-comment">// 现在可以消费数据event了</span>
})

<span class="hljs-comment">// Promise</span>
<span class="hljs-keyword">var</span> promise = <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>(<span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">resolve, reject</span>) </span>{
    setTimeout(<span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params"></span>) </span>{
        <span class="hljs-keyword">var</span> data = <span class="hljs-number">42</span>;
        resolve(data);
    }, <span class="hljs-number">3000</span>);
});
promise.then(<span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">data</span>) </span>{
    <span class="hljs-comment">// 可以开始消费数据data了</span>
});

<span class="hljs-comment">// RxJS</span>
<span class="hljs-keyword">var</span> observable = Rx.Observable.fromEvent(button, <span class="hljs-string">'click'</span>);
observable.subscribe(<span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">event</span>) </span>{
    <span class="hljs-comment">// 可以开始消费数据event了</span>
})
</code></pre>
<h2>基本概念</h2>
<p><strong>除了拥有异步处理能力，RxJS还提供了对生产的数据进行常见操作的接口，以及提供多种触发回调的策略（比如同步、异步等）</strong>。理解RxJS的概念有利于更好地利用它，接下来介绍RxJS所包含的基本概念。</p>
<h3>Observable</h3>
<p>使用RxJS的第一步就是得到一个Observable，可以把它理解为数据流，<strong>它可以向订阅者（即消费者）推送多个数据</strong>。Observable分两种，分别是Cold Observable和Hot Observable。Cold Observable的特点是每个订阅者独享该Observable推送的数据序列，并且数据生产过程是在添加订阅者（比如执行了<code>subscribe</code>方法）的时候才开始启动的；而Hot Observable的订阅者则共享该Observable推送的数据序列，启动数据生产过程一般通过特定方法（比如调用<code>connect</code>方法）来启动。</p>
<p>以下代码展示Cold Observable中订阅者独享数据序列的特点：</p>
<pre><code class="language-js"><span class="hljs-keyword">var</span> coldObservable = Rx.Observable.create(<span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">observer</span>) </span>{
  observer.next({
      <span class="hljs-attr">a</span>: <span class="hljs-number">1</span>,
      <span class="hljs-attr">b</span>: <span class="hljs-number">2</span>
  });
});
<span class="hljs-comment">// 订阅的时候给每一个订阅者形成单独一份返回值序列，并且包含之前的返回值</span>
coldObservable.subscribe({
    <span class="hljs-attr">next</span>: <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">data</span>) </span>{
        <span class="hljs-built_in">console</span>.log(data.a); <span class="hljs-comment">// 输出 1</span>
        data.a = <span class="hljs-number">3</span>;
    }
});
coldObservable.subscribe({
    <span class="hljs-attr">next</span>: <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">data</span>) </span>{
        <span class="hljs-comment">// 输出 1，上一个订阅者的修改不会影响到</span>
        <span class="hljs-built_in">console</span>.log(data.a);
        data.a = <span class="hljs-number">3</span>;
    }
});
</code></pre>
<p>以下代码展示Hot Observable中订阅者共享数据序列的特点：</p>
<pre><code class="language-js"><span class="hljs-keyword">var</span> hotObservable = Rx.Observable.create(<span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">observer</span>) </span>{
  observer.next({
      <span class="hljs-attr">a</span>: <span class="hljs-number">1</span>,
      <span class="hljs-attr">b</span>: <span class="hljs-number">2</span>
  });
}).publish();;
hotObservable.subscribe({
    <span class="hljs-attr">next</span>: <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">data</span>) </span>{
        <span class="hljs-comment">// 输出 1</span>
        <span class="hljs-built_in">console</span>.log(data.a);
        <span class="hljs-comment">// 修改属性值</span>
        data.a = <span class="hljs-number">3</span>;
    }
});
hotObservable.subscribe({
    <span class="hljs-attr">next</span>: <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">data</span>) </span>{
        <span class="hljs-comment">// 输出修改后的属性值 3，说明是共享Observable推送的数据序列</span>
        <span class="hljs-built_in">console</span>.log(data.a);
        data.a = <span class="hljs-number">3</span>;
    }
});
<span class="hljs-comment">// 开始形成返回值序列，并且每个订阅者共享这个序列</span>
hotObservable.connect();
</code></pre>
<h3>Observer</h3>
<p>Observer是一个包含回调函数的对象（可包含<code>next</code>、<code>complete</code>、<code>error</code>三种不同状态的回调），称之为订阅者，比较好理解。例如上面在调用Observable的<code>subscribe</code>方法时所传入的参数，就是一个Observer。</p>
<h3>Subscription</h3>
<p>Subscription表示给Observable添加一个订阅者这个动作对象。对于Cold Observable，<code>subscribe</code>方法会返回一个Subscription，并且会启动推送数据的过程，同时这个动作对象通过调用自己的<code>unsubscribe</code>进行取消。</p>
<h3>Subject</h3>
<p>Subject跟Hot Observable类似，区别是Subject也可以看作一个Observer，同时在创建Subject对象时就默认启动了数据生产过程。因此，Subject可以通过<code>next</code>向其订阅者发送数据：</p>
<pre><code class="language-js"><span class="hljs-keyword">var</span> subject = <span class="hljs-keyword">new</span> Rx.Subject();
subject.subscribe({
    <span class="hljs-attr">next</span>: <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">data</span>) </span>{
        <span class="hljs-built_in">console</span>.log(data); <span class="hljs-comment">// 输出1和2</span>
    }
});
subject.next(<span class="hljs-number">1</span>);
subject.subscribe({
    <span class="hljs-attr">next</span>: <span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">data</span>) </span>{
        <span class="hljs-built_in">console</span>.log(data); <span class="hljs-comment">// 只输出2</span>
    }
});
subject.next(<span class="hljs-number">2</span>);
</code></pre>
<h3>Operator</h3>
<p>Operator是操作异步数据的方法，返回值是另一个Observable对象，可链式调用。常见的操作有处理数据（<code>map</code>）、过滤数据（<code>filter</code>）、累积数据（<code>reduce</code>）等。例如以下示例代码：</p>
<pre><code class="language-js">Rx.Observable.of(<span class="hljs-number">0</span>, <span class="hljs-number">1</span>, <span class="hljs-number">2</span>, <span class="hljs-number">3</span>)
    .map(<span class="hljs-function">(<span class="hljs-params">n</span>) =&gt;</span> n * n) <span class="hljs-comment">// 每项取平方</span>
    .filter(<span class="hljs-function">(<span class="hljs-params">n</span>) =&gt;</span> n &gt; <span class="hljs-number">0</span>) <span class="hljs-comment">// 取出平方后大于0的项</span>
    .reduce(<span class="hljs-function">(<span class="hljs-params">acc, curr</span>) =&gt;</span> acc + curr, <span class="hljs-number">0</span>) <span class="hljs-comment">// 累加每一项</span>
    .subscribe(<span class="hljs-function">(<span class="hljs-params">sum</span>) =&gt;</span> <span class="hljs-built_in">console</span>.log(sum)); <span class="hljs-comment">// 输出 14，等于1*1 + 2*2 + 3*3</span>
</code></pre>
<h3>Scheduler</h3>
<p>Scheduler指触发回调的策略，是一个对象。这里的策略，控制了触发回调的时机。比如把触发放到下一次事件循环里面去。以下代码展示了使用<code>async</code>策略的运行结果（重点在于使用<code>async</code>策略会异步触发回调，）：</p>
<pre><code class="language-js"><span class="hljs-keyword">var</span> observable = Rx.Observable.create(<span class="hljs-function"><span class="hljs-keyword">function</span> (<span class="hljs-params">observer</span>) </span>{
    <span class="hljs-built_in">console</span>.log(<span class="hljs-string">'a'</span>);
    observer.next(<span class="hljs-number">1</span>);
    <span class="hljs-built_in">console</span>.log(<span class="hljs-string">'b'</span>);
    observer.complete();
    <span class="hljs-built_in">console</span>.log(<span class="hljs-string">'c'</span>);
}).observeOn(Rx.Scheduler.async);

<span class="hljs-built_in">console</span>.log(<span class="hljs-string">'before'</span>);
observable.subscribe({
  <span class="hljs-attr">next</span>: <span class="hljs-function">(<span class="hljs-params">x</span>) =&gt;</span> <span class="hljs-built_in">console</span>.log(x),
  <span class="hljs-attr">complete</span>: <span class="hljs-function"><span class="hljs-params">()</span> =&gt;</span> <span class="hljs-built_in">console</span>.log(<span class="hljs-string">'done'</span>),
});
<span class="hljs-built_in">console</span>.log(<span class="hljs-string">'after'</span>);
</code></pre>
<p>输出顺序如下：</p>
<pre><code>before
a
b
c
after
1
done
</code></pre>
<p>由于目前JS解释器都是单线程运行JS代码，Scheduler的作用局限于提供异步等策略。但是在Rx.NET、RxJava等Rx
的实现中，Scheduler有非常大的作用，比如可以将耗时的任务放到新的线程中，从而提高主线程的响应速度。对这部分感兴趣的同学可以去阅读Rx的相关实现，比如Rx.NET、RxJava。所有语言和平台的Rx实现在<a href="http://reactivex.io/languages.html">这里</a>可以看到。</p>
<h2>总结</h2>
<p>本文主要介绍了RxJS中涉及到的基本概念，并提供代码示例。理解这些概念是使用好RxJS的基础，同时RxJS包含了丰富的Operator。在解决实际问题时，对这些Operator的理解有利于找到快速高效的解决方法。如果你打算继续深入学习RxJS的话，建议从Operator入手，并结合实际问题。</p>
<p>接下来的问题是：为什么要使用RxJS？RxJS适合解决哪些问题？这些是下一篇RxJS文章所涉及到的内容，敬请期待。</p>


</body>
</html>
